package com.ufo.dis.service.entity;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;

public class HdfsDAO {

    // HDFS访问地址
    private static final String HDFS = "hdfs://localhost:9000";
    // hdfs路径
    private String hdfsPath;
    // Hadoop系统配置
    private Configuration conf;

    public HdfsDAO(Configuration conf) {
        this(HDFS, conf);
    }

    public HdfsDAO(String hdfs, Configuration conf) {

        this.hdfsPath = hdfs;
        this.conf = conf;
    }

    public static String getHdfs() {
        return HDFS;
    }


    // 加载Hadoop配置文件
    public static JobConf config() {
        JobConf conf = new JobConf(HdfsDAO.class);
        System.out.println("配置中...");
        conf.setJobName("HdfsDAO");
        conf.addResource("classpath:/hadoop-config/core-site.xml");
        conf.addResource("classpath:/hadoop-config/hdfs-site.xml");
        conf.addResource("classpath:/hadoop-config/mapred-site.xml");
        System.out.println("配置完成...");
        return conf;
    }

    // 在根目录下创建文件夹
    public void mkdirs(String folder) throws IOException {
        Path path = new Path(folder);
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        if (!fs.exists(path)) {
            fs.mkdirs(path);
            System.out.println("Create: " + folder);
        }
        fs.close();
    }

    // 某个文件夹的文件列表
    public FileStatus[] ls(String folder) throws IOException {
        Path path = new Path(folder);
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        FileStatus[] list = fs.listStatus(path);
        System.out.println("ls: " + folder);
        System.out.println("==========================================================");
        if (list != null)
            for (FileStatus f : list) {
                System.out.printf("%s, folder: %s, 大小: %dK\n", f.getPath().getName(), (f.isDirectory() ? "目录" : "文件"),
                        f.getLen() / 1024);
            }
        System.out.println("==========================================================");
        fs.close();

        return list;
    }

    // 将文件上传到hdfs
    public void upload1(String local, String remote) throws IOException {
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        // remote---/用户/用户下的文件或文件夹
        fs.copyFromLocalFile(new Path(local), new Path(remote));
        System.out.println("copy from: " + local + " to " + remote);
        fs.close();
    }
//上传第二种
    public void upload2(String fileName, InputStream in) throws IllegalArgumentException, IOException, InterruptedException, URISyntaxException {

        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        FSDataOutputStream out = fs.create(new Path(fileName));
        IOUtils.copy(in, out);
        fs.close();
    }



    //上传第三种
    public void upload3(String sources ,  String dests) throws IllegalArgumentException, IOException, InterruptedException, URISyntaxException {
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        Path source = new Path(sources);
        Path dest = new Path(dests);
        fs.copyFromLocalFile(true, false, source, dest);
        fs.close();
    }

    // 删除文件或文件夹
    public void rmr(String folder) throws IOException {
        Path path = new Path(folder);
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        fs.deleteOnExit(path);
        System.out.println("Delete: " + folder);
        fs.close();
    }

    //移动目录或文件到路径
    public boolean mv(String path, String newPath) throws IOException {
        boolean result = false;
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        try {

            if (fs.exists(new Path(newPath))) {
                result = fs.rename(new Path(path), new Path(newPath));
                result=true;
            } else {
//                LOGGER.warn("HDFS上目录： {} 被占用！",newPath);
            }
        } catch (Exception e) {
//            LOGGER.error("移动HDFS上目录：{} 失败！", path, e);
        } finally {
            fs.close();
        }
        System.out.println("mv: " + path+"  to  "+newPath);
        return result;
    }


}